package com.systar.activity.monitor;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.google.common.base.Strings;
import com.google.common.eventbus.EventBus;
import com.systar.activity.IActivable;
import com.systar.activity.events.IActivityEvent;
import com.systar.activity.parse.FileParser;
import com.systar.activity.parse.ParserFactory;
import com.systar.activity.process.ActivityStack;
import com.systar.activity.process.online.HttpNotifier;
import com.systar.activity.process.online.ZookeeperNotifier;
import com.systar.activity.state.FileActivity;

public class ActivityManager
{
	private final List<ActivitySource> sources;
	private ExecutorCompletionService<Integer> executorCompletionService;
	private ExecutorService executorService;
	private final ActivityStack eventStack;
	private IActivable onlineNotifier;
	private volatile boolean closing = false;

	public ActivityManager(Properties props, boolean checkCompletion)
	{
		sources = new ArrayList<ActivitySource>();
		eventStack = new ActivityStack();

		initialize(props, checkCompletion);
	}

	public ActivityStack createEventDelegation()
	{
		return eventStack.createDelegation();
	}
	
	public Iterable<ActivitySource> getSources()
	{
		return sources;
	}

	private void initialize(Properties props, boolean checkCompletion)
	{
		// create sources
		int i = 1;
		while (true)
		{
			String keyName = String.format("source%d_name", i);
			String keyLocation = String.format("source%d_location", i);
			if (!props.containsKey(keyName) || !props.containsKey(keyLocation))
				break;
			String valueName = props.getProperty(keyName);
			String valueLocation = props.getProperty(keyLocation);
			sources.add(new ActivitySource(this, valueName, valueLocation));
			i++;
		}

		// parsing threads
		int nbThreads = Integer.parseInt(props.getProperty("ParseThreads", "2"));
		executorService = Executors.newFixedThreadPool(nbThreads);
		if (checkCompletion)
			executorCompletionService = new ExecutorCompletionService<Integer>(executorService);
		
		// Web server notifications
		String zookeeperURL = props.getProperty("zookeeperURL");
		String serverURL = props.getProperty("serverURL");
		if (!Strings.isNullOrEmpty(zookeeperURL))
		{
			onlineNotifier = new ZookeeperNotifier(this, zookeeperURL);
		}
		else if (!Strings.isNullOrEmpty(serverURL))
		{
			onlineNotifier = new HttpNotifier(this, serverURL);
		}
	}

	public boolean isClosing()
	{
		return closing;
	}

	public void parseFile(final FileActivity activity, final EventBus eventBus)
	{
		FileParser parser = ParserFactory.getParser(activity, eventBus);
		if (parser != null)
		{
			if (executorCompletionService != null)
				executorCompletionService.submit(parser);
			else
				executorService.submit(parser);
			
			System.out.println("parsing file: "+activity.getFile().getName());
		}
	}

	public void handleDisplayEvent(IActivityEvent event)
	{
		if (isClosing())
			return;
		eventStack.add(event);
	}

	public List<IActivityEvent> getDisplayEvents()
	{
		return eventStack.get();
	}

	public IActivable getOnlineNotifier()
	{
		return onlineNotifier;
	}

	public void start()
	{
		if (onlineNotifier != null)
		{
			onlineNotifier.activate();
		}
		for (ActivitySource source : getSources())
		{
			source.activateObserver();
		}
	}

	public void shutdown()
	{
		closing = true;
		executorService.shutdown();
		try
		{
			executorService.awaitTermination(1000, TimeUnit.MILLISECONDS);
		}
		catch (InterruptedException e)
		{
			// skip
		}
		for (ActivitySource source : getSources())
		{
			source.deactivateObserver();
		}
		if (onlineNotifier != null)
		{
			onlineNotifier.deactivate();
		}
	}

	public void waitForExecutors() throws InterruptedException
	{
		if (executorCompletionService != null)
			executorCompletionService.take();
	}
}
